package com.jason.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Logger;

import com.jason.nio.util.RunExecutor;

public class NioIoProcesser {
	
	private Logger log = Logger.getLogger("NioIoProcesser");
	
	private Object lock = new Object();
	
	private final Queue<NioSession> newSessions = new ConcurrentLinkedQueue<NioSession>();
	
	private Selector selector;
	
	private Worker worker;
	
	private RunExecutor executor;
	
	public NioIoProcesser(RunExecutor executor){
		this.executor = executor;
	}
	
	
	public void addNew(NioSession session) throws IOException{
		log.info("new session add!");
		newSessions.add(session);
		startWorker();
	}
	
	public void startWorker() throws IOException{
		synchronized(lock){
			if(worker == null){
				selector = Selector.open();
				worker = new Worker();
				executor.execute(worker);
			}
			selector.wakeup();
		}
	}
	
	public void doAddNew(){
		for(;;){
			NioSession session = newSessions.poll();
			if(session == null){
				break;
			}
			SocketChannel sc = session.getCh();
			try{
				sc.configureBlocking(false);
				sc.register(selector, SelectionKey.OP_READ,session);
			}catch(Exception e){
				e.printStackTrace();
			}
		}
	}
	
	public void process(Set<SelectionKey> keys){
		
		for(SelectionKey key:keys){
			NioSession session = (NioSession)key.attachment();
			if(key.isReadable()){
				read(session);
			}
		}
	}
	
	public void read(NioSession session){
		ByteBuffer buff = ByteBuffer.allocate(session.getReadBufferSize());
		SocketChannel sc = session.getCh();
		try{
			int readBytes = 0;
			int ret;
			while((ret=sc.read(buff)) >0){
				readBytes +=ret;
			}
			buff.flip();
			if(readBytes >0){
				System.out.println("read bytes:" +readBytes);
			}
		}catch(Exception e){
			e.printStackTrace();
		}
	}
	
	
	private class Worker implements Runnable{
		
		public void run(){
			Selector selector = NioIoProcesser.this.selector;
			for(;;){
				try{
					int key = selector.select();
					doAddNew();
					if(key >0){
						process(selector.selectedKeys());
					}
				}catch(Exception e){
					e.printStackTrace();
				}
			}
		}
	}

}
